package net.bwie.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * 仿双十一实时大屏统计：第2步、对实时采集数据进行处理，结果存储MySQL表
 * @author xuanyu
 * @date 2025/10/20
 */
public class _04TmallRealtimeDemo03 {

	public static void main(String[] args) throws Exception{
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// 2. 数据源-source
		KafkaSource<String> source = KafkaSource.<String>builder()
			.setBootstrapServers("node101:9092,node102:9092,node103:9092")
			.setTopics("dwd_order")
			.setGroupId("my-group-1")
			.setStartingOffsets(OffsetsInitializer.earliest())
			.setValueOnlyDeserializer(new SimpleStringSchema())
			.build();
		DataStreamSource<String> stream = env.fromSource(
			source, WatermarkStrategy.noWatermarks(), "Kafka Source"
		);
		//stream.print("kafka");

		// todo 总销售额实时统计，针对order_info表数据处理
		// 3-1-1. 获取order_info表数据
		SingleOutputStreamOperator<String> stream3_11 = stream.filter(
			new FilterFunction<String>() {
				@Override
				public boolean filter(String value) throws Exception {
					// value 属于json格式数据，获取table字段值，两次json数据解析
					JSONObject jsonObject = JSON.parseObject(value);
					// 再次解析source字段
					JSONObject sourceJsonObject = jsonObject.getJSONObject("source");
					// 获取字段值
					String tableValue = sourceJsonObject.getString("table");
					// 判断
					return "order_detail".equals(tableValue);
				}
			}
		);
		// 3-1-2. 提取字段值
		SingleOutputStreamOperator<Tuple2<String, Double>> stream3_12 = stream3_11.map(
			new MapFunction<String, Tuple2<String, Double>>() {
				@Override
				public Tuple2<String, Double> map(String value) throws Exception {
					// value 属于json格式数据
					JSONObject jsonObject = JSON.parseObject(value);
					// 解析after值
					JSONObject afterJsonObject = jsonObject.getJSONObject("after");
					// 获取字段值
					String skuId = afterJsonObject.getString("sku_id");
					Double orderPrice = afterJsonObject.getDouble("order_price");
					Integer skuNum = afterJsonObject.getInteger("sku_num");
					// 返回二元组
					return Tuple2.of(skuId, orderPrice * skuNum);
				}
			}
		);
		// 3-1-3. 分组
		KeyedStream<Tuple2<String, Double>, String> stream3_13 = stream3_12.keyBy(
			new KeySelector<Tuple2<String, Double>, String>() {
				@Override
				public String getKey(Tuple2<String, Double> value) throws Exception {
					return value.f0;
				}
			}
		);
		// 3-1-4. 聚合
		SingleOutputStreamOperator<Tuple2<String, Double>> stream3_14 = stream3_13.sum(1);
		//stream3_14.print();

		// 保存MySQL表中
		/*
-- 使用数据库
USE gmall;
CREATE TABLE IF NOT EXISTS tmall_report(
      tmall_type VARCHAR(255) PRIMARY KEY ,
      total_amount double
);

-- 插入数据
REPLACE INTO gmall.tmall_report(tmall_type, total_amount) VALUES (?, ?) ;
		 */
		SinkFunction<Tuple2<String, Double>> jdbcSink = JdbcSink.sink(
			"REPLACE INTO gmall.tmall_report(tmall_type, total_amount) VALUES (?, ?)",
			(statement, tuple) -> {
				statement.setObject(1, tuple.f0);
				statement.setObject(2, tuple.f1);
			},
			JdbcExecutionOptions.builder()
				.withBatchSize(1000)
				.withBatchIntervalMs(200)
				.withMaxRetries(5)
				.build(),
			new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
				.withUrl("jdbc:mysql://node101:3306/gmall")
				.withDriverName("com.mysql.jdbc.Driver")
				.withUsername("root")
				.withPassword("123456")
				.build()
		);
		stream3_14.addSink(jdbcSink);


		// 5. 触发执行-execute
		env.execute("TmallRealtimeDemo01") ;
	}

}
